#include "threadPool.h"
#include <iostream>
#include <cstring>
#include <unistd.h>
using namespace std;

#define DEFAULT_MIN_THREADS 3
#define DEFAULT_MAX_THREADS 8
#define DEFAULT_MAX_QUEUE_CAPACITY 20

#define SLEEP_INTERVAL 5
#define INCREASE_NUM 2
#define DECREASE_NUM 2

/* 线程退出 */
void ThreadPool::exitThread(pthread_t tid)
{
    for (int idx = 0; idx < this->m_maxThreadNum; idx++)
    {
        if (m_threadId[idx] == tid)
        {
            m_threadId[idx] = 0;

            /* 线程退出 */
            pthread_exit(NULL);
        }
    }
}
/* 管理者线程 */
void *m_managerThreadFunc(void *arg)
{
    /* 线程分离 */
    pthread_detach(pthread_self());

    ThreadPool *pool = (ThreadPool *)arg;

    /* 调度. */
    while (1)
    {
        /* 休眠 5s */
        sleep(SLEEP_INTERVAL);

        /* 加锁 */
        pthread_mutex_lock(&pool->m_mutex);
        int aliveThreadNum = pool->m_aliveThreadNum;
        int queueSize = pool->m_queueSize;
        /* 解锁*/
        pthread_mutex_unlock(&pool->m_mutex);

        /* 加锁 */
        pthread_mutex_lock(&(pool->m_busyMutex));
        int busyThreadNum = pool->m_busyThreadNum;

        /*解锁*/
        pthread_mutex_unlock(&pool->m_busyMutex);

        // cout << "queueSize:" << queueSize << ", aliveThreadNum:" << aliveThreadNum << ", busyThreadNum:" << busyThreadNum << endl;

        /* 1.什么时候需要增加线程? */
        int ret = 0;
        if (queueSize > (aliveThreadNum - busyThreadNum) && (aliveThreadNum < pool->m_maxThreadNum))
        {
            int addCnt = 0;
            /* 加锁 */
            pthread_mutex_lock(&(pool->m_mutex));

            for (int idx = 0; idx < pool->m_maxThreadNum && addCnt < INCREASE_NUM; idx++)
            {
                /* 探测 */
                if (pool->m_threadId[idx] == 0)
                {
                    ret = pthread_create(&(pool->m_threadId[idx]), NULL, workerThreadFunc, pool);
                    if (ret == -1)
                    {
                        perror("workerThread create error.");
                        _exit(-1);
                    }
                    printf("%lu join thread pool.\n", pool->m_threadId[idx]);

                    addCnt++;
                    /* 存活的线程数 +1 */
                    (pool->m_aliveThreadNum)++;
                }
            }

            /* 解锁 */
            pthread_mutex_unlock(&(pool->m_mutex));
        }

        /* 2.什么时候减少线程池的线程? */
        if ((queueSize < (aliveThreadNum - busyThreadNum)) && (aliveThreadNum > pool->m_minThreadNum))
        {
            /* 减少线程 加锁 */
            pthread_mutex_lock(&(pool->m_mutex));

            /* 减少的线程数 */
            pool->exitNum = DECREASE_NUM;

            for (int idx = 0; idx < DECREASE_NUM; idx++)
            {
                pthread_cond_signal(&(pool->m_queueNotEmpty));
            }

            /* 解锁  */
            pthread_mutex_unlock(&(pool->m_mutex));
        }
    }
}

/* 工作者线程 */
void *workerThreadFunc(void *arg)
{
    /* 线程分离 */
    pthread_detach(pthread_self());

    ThreadPool *pool = (ThreadPool *)arg;
    while (true)
    {
        /* 加锁 */
        pthread_mutex_lock(&pool->m_mutex);
        while (pool->m_queueSize == 0)
        {
            /* 让出CPU 休眠10ms */
            usleep(10000);
            /* 等一个信号 */
            pthread_cond_wait(&(pool->m_queueNotEmpty), &(pool->m_mutex)); // 1.阻塞等待 2.解锁 3.加锁

            if (pool->exitNum > 0)
            {
                pool->exitNum--;
                if (pool->m_aliveThreadNum > pool->m_minThreadNum)
                {
                    /* 存活的线程数 -1 */
                    (pool->m_aliveThreadNum)--;
                    /*  解锁 */
                    pthread_mutex_unlock(&(pool->m_mutex));

                    printf("%lu exit thread pool...\n", pthread_self());

                    /* 线程退出 */
                    pool->exitThread(pthread_self());
                }
            }
        }
        /* 程序到这个地方说明:what? */
        Task_t newTask = pool->m_queue[pool->m_queueFront];

        /* 将队列的队头向后迭代. */
        pool->m_queueFront = (pool->m_queueFront + 1) % pool->m_queueCapacity;
        /* 队列的元素个数 -1 */
        pool->m_queueSize--;
        /* 解锁 */
        pthread_mutex_unlock(&(pool->m_mutex));

        pthread_mutex_lock(&(pool->m_busyMutex));
        /* 忙碌的线程数 +1 */
        pool->m_busyThreadNum++;
        pthread_mutex_unlock(&(pool->m_busyMutex));
        /* 线程开始执行任务 */
        newTask.function(newTask.arg);

        pthread_mutex_lock(&(pool->m_busyMutex));
        /* 忙碌的线程数 -1 */
        pool->m_busyThreadNum--;
        pthread_mutex_unlock(&(pool->m_busyMutex));

        /* 发信号 */
        pthread_cond_signal(&pool->m_queueNotFull);
    }
}

/* 构造函数 */
ThreadPool::ThreadPool(int minThread, int maxThread, int maxQueueCapacity)
{
    if (minThread <= 0 || maxThread <= 0 || minThread >= maxThread)
    {
        minThread = DEFAULT_MIN_THREADS;
        maxThread = DEFAULT_MAX_THREADS;
    }

    if (maxQueueCapacity <= 0)
    {
        maxQueueCapacity = DEFAULT_MAX_QUEUE_CAPACITY;
    }

    /* 队列 */
    this->m_queueCapacity = maxQueueCapacity;
    /* 队列的队头 */
    this->m_queueFront = m_queueTail = 0;
    /* 队列的队尾 */
    m_queueSize = 0;
    this->m_queue = new Task_t[maxQueueCapacity];
    /* 清除脏数据 */
    memset(this->m_queue, 0, sizeof(Task_t) * maxQueueCapacity);

    /* 工作线程Id */
    m_threadId = new pthread_t[maxThread];
    /* 清除脏数据 */
    memset(m_threadId, 0, sizeof(pthread_t) * maxThread);

    this->m_minThreadNum = minThread;
    this->m_maxThreadNum = maxThread;
    this->m_aliveThreadNum = 0;
    this->m_busyThreadNum = 0;

    int ret = 0;
    /* 创建管理者线程 */
    ret = pthread_create(&m_managerThreadId, NULL, m_managerThreadFunc, this);
    if (ret == -1)
    {
        perror("managerThread create error.");
        _exit(-1);
    }

    /* 创建线程 */
    for (int idx = 0; idx < minThread; idx++)
    {
        ret = pthread_create(&m_threadId[idx], NULL, workerThreadFunc, this);
        if (ret == -1)
        {
            perror("workerThread create error.");
            _exit(-1);
        }
        /* 存活的线程数++ */
        (this->m_aliveThreadNum)++;
    }

    /* 初始化锁 */
    pthread_mutex_init(&m_mutex, NULL);
    pthread_mutex_init(&m_busyMutex, NULL);
    /* 初始化条件变量 */
    pthread_cond_init(&m_queueNotEmpty, NULL);
    pthread_cond_init(&m_queueNotFull, NULL);

    /* 初始化为0 */
    this->exitNum = 0;
}

/*添加任务 */
void ThreadPool::addTask(void *(*function)(void *arg), void *arg)
{
    /* 将任务添加到任务队列 */

    /* 加锁 */
    pthread_mutex_lock(&m_mutex);

    while (this->m_queueCapacity == this->m_queueSize)
    {
        pthread_cond_wait(&m_queueNotFull, &m_mutex);
    }
    /* 程序执行到这里:将任务添加到队列里面 */
    m_queue[m_queueTail].function = function;
    m_queue[m_queueTail].arg = arg;

    m_queueTail = (m_queueTail + 1) % m_queueCapacity;

    /* 任务数量 +1 */
    this->m_queueSize++;

    /* 解锁 */
    pthread_mutex_unlock(&m_mutex);
    /* 发信号 */
    pthread_cond_signal(&m_queueNotEmpty);
}

ThreadPool::~ThreadPool()
{
    if (m_threadId != NULL)
    {
        delete[] m_threadId;
    }
    if (m_queue != NULL)
    {
        delete[] m_queue;
    }
    /* 释放锁 */
    pthread_mutex_destroy(&m_mutex);
    pthread_mutex_destroy(&m_busyMutex);

    pthread_cond_destroy(&m_queueNotEmpty);
    pthread_cond_destroy(&m_queueNotFull);
}


